package cn.spark.study.core;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

/**
 * 二次排序
 * 实现自定义key,要实现Ordered接口和Serializable接口，在key中实现自己对多个列的排序算法
 *
 * @author jun.zhang6
 * @date 2020/11/6
 */
public class SecondarySort {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf()
                .setAppName("SecondarySort")
                .setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        JavaRDD<String> linesRdd = sc.textFile("C:\\Users\\jun.zhang6\\Desktop\\spark.txt");

        JavaPairRDD<SecondarySortKey, String> pairsRdd = linesRdd.mapToPair(new PairFunction<String, SecondarySortKey, String>() {
            @Override
            public Tuple2<SecondarySortKey, String> call(String line) throws Exception {
                String[] splitedLines = line.split(" ");
                SecondarySortKey secondarySortKey = new SecondarySortKey(Integer.valueOf(splitedLines[0]), Integer.valueOf(splitedLines[1]));
                return new Tuple2<SecondarySortKey, String>(secondarySortKey, line);
            }
        });

        JavaPairRDD<SecondarySortKey, String> sortedRdd = pairsRdd.sortByKey(false);

        JavaRDD<String> sortedLines = sortedRdd.map(new Function<Tuple2<SecondarySortKey, String>, String>() {
            @Override
            public String call(Tuple2<SecondarySortKey, String> tuple) throws Exception {
                return tuple._2();
            }
        });

        sortedLines.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });

        sc.close();

    }
}
